package com.jaymin.flink.datastream.source;

import com.jaymin.flink.datastream.transformation.Access;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author jaymin
 * @since 2024/4/5 1:24
 */
public class SourceApp {

    public static void main(String[] args) throws Exception {
        // 创建上下文，推荐使用
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 本地上下文
        LocalStreamEnvironment localStreamEnvironment = StreamExecutionEnvironment.createLocalEnvironment();

        // 设置并行度
        StreamExecutionEnvironment.createLocalEnvironment(2);

        StreamExecutionEnvironment.createLocalEnvironment(new Configuration());

        StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

//        StreamExecutionEnvironment.createRemoteEnvironment();

//        test01(env);
//        test02(env);
        test03(env);
        env.execute("TEST");
    }

    private static void test03(StreamExecutionEnvironment env) {
        DataStreamSource<Student> accessDataStreamSource = env.addSource(new MysqlSource());
        int parallelism = accessDataStreamSource.getParallelism();
        System.out.println(parallelism);
        accessDataStreamSource.print();
    }

    private static void test02(StreamExecutionEnvironment env) {
        DataStreamSource<Access> accessDataStreamSource = env.addSource(new AccessSourceV2());
        int parallelism = accessDataStreamSource.getParallelism();
        System.out.println(parallelism);
        accessDataStreamSource.print();
    }

    private static void test01(StreamExecutionEnvironment env) {
        DataStreamSource<Access> accessDataStreamSource = env.addSource(new AccessSource());
        int parallelism = accessDataStreamSource.getParallelism();
        accessDataStreamSource.print();

    }
}
